package io.nio;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 使用线程池处理客户端与服务端的io操作
 */
public class PoolServer {
    // 固定线程数连接池
    private ExecutorService pool = Executors.newFixedThreadPool(50);
    private Selector selector;

    public static void main(String[] args) {
        PoolServer poolServer = new PoolServer();
        try {
            poolServer.initServer(9092);
            poolServer.listen();
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    /**
     * 轮询 selector ,处理连接事件，并将读事件交给线程池执行
     *
     * @throws IOException
     */
    private void listen() throws IOException {
        while (true) {
            //阻塞方法，只有当所有通道中有就绪的通道才厚向下继续执行代码
            selector.select();
            System.out.println("selector 轮询完成。。。。。。。");
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                iterator.remove();
                // 处理连接事件
                if (key.isAcceptable()) {
                    ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                    SocketChannel sc = ssc.accept();
                    sc.configureBlocking(false);
                    sc.register(key.selector(), SelectionKey.OP_READ);
                } else if (key.isReadable()) {
                    // 当有io 数据是可以读取的时候 将读取操作交给线程池去启动一个单独的线程去处理
                    // key.interestOps() &~(SelectionKey.OP_READ) 移除读事件监听
                    key.interestOps(key.interestOps() & ~(SelectionKey.OP_READ));
                    pool.execute(new ThreadHandlerChannel(key));
                }
            }

        }
    }


    /**
     * 初始化服务端
     *
     * @param port
     * @throws IOException
     */
    private void initServer(int port) throws IOException {
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        serverSocketChannel.socket().bind(new InetSocketAddress(port));
        this.selector = Selector.open();
        // 向selector 注册 客户端连接事件
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        System.out.println("服务端启动完成！");
    }
}

class ThreadHandlerChannel extends Thread {
    private SelectionKey key;

    public ThreadHandlerChannel(SelectionKey key) {
        this.key = key;
    }

    @Override
    public void run() {
        SocketChannel sc = (SocketChannel) key.channel();
        // 创建一个缓冲区
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        // 创建一个字节数组输出流，用于写出数据
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        int size = 0;
        try {
            while (true) {
                if ((size = sc.read(buffer)) > 0) {
                    buffer.flip();
                    // 写到输出流里面
                    baos.write(buffer.array(), 0, size);
                    buffer.clear();
                }
                // 流关闭后不可以在进行读写，但是其他的操作还是可以做的
                baos.close();
                // 获取字节输出流中的字节数组
                byte[] context = baos.toByteArray();
                // 创建一个缓冲区，并将从通道中读到的字节数组写入到缓冲区里面
                ByteBuffer writeBuf = ByteBuffer.allocate(context.length);
                writeBuf.put(context);
                writeBuf.flip();
                sc.write(writeBuf);
                if (size == -1) {
                    // 客户端关闭连接后，关闭通道，并结束循环
                    sc.close();
                    break;
                } else {
                    // 注册读事件监听
                    key.interestOps(key.interestOps() | SelectionKey.OP_READ);
                    key.selector().wakeup();
                }
            }
        } catch (IOException e) {
            e.printStackTrace();

        }


    }
}
